#include "server.h"

static struct
{/*{{{*/
    unsigned int thrd_id;
    EP ep;
    EP_SLOT server_slot;
    EP_SLOT *worker_slot;
}/*}}}*/
listener_ctl;

static int _conn(struct ep_active_slot *active_slot)
{/*{{{*/
#define _MAX_CONN_NUM        100
    SOCKET new_sock=active_slot->_u._conn.sock;
    char msg[100];
    int msg_size;
    int worker_idx;
    struct _worker_info *worker=NULL;
    int idle_num;
    int idle[cnfg_info.threads];
    int min_busy_conn_num;
    int min_busy_idx;
    int res;

    pthread_mutex_lock(&stat_info._worker_.lock);

    idle_num=0;
    min_busy_conn_num=_MAX_CONN_NUM;
    min_busy_idx=-1;
    for(worker_idx=0; worker_idx<cnfg_info.threads; ++worker_idx)
    {
        worker=&stat_info._worker_.worker[worker_idx];
        if(worker->conn_num == 0)
        {   idle[idle_num++]=worker_idx;   }
        else
        {
            if(min_busy_conn_num > worker->conn_num)
            {
                min_busy_conn_num=worker->conn_num;
                min_busy_idx=worker_idx;
            }
        }
    }
    if(idle_num > 0) // 选取无连接的worker
    {
        worker_idx=idle[random_int(0, idle_num-1)];
#ifdef _MY_DEBUG
        errlog("%s: choose idle thread(%d)",
                __func__, worker_idx);
#endif
    }
    else if(min_busy_conn_num < _MAX_CONN_NUM) // 选取连接最少的worker
    {
        worker_idx=min_busy_idx;
    }
    else
    {
        errlog("%s: worker full, BAD", __func__);
        pthread_mutex_unlock(&stat_info._worker_.lock);
        sock_close(new_sock);
        return(-1);
    }
    ++stat_info._worker_.worker[worker_idx].conn_num;
#ifdef _MY_DEBUG
    errlog("%s: worker(%d) conn_num:%d", __func__, worker_idx,
            stat_info._worker_.worker[worker_idx].conn_num);
#endif

    pthread_mutex_unlock(&stat_info._worker_.lock);

    CLR_BUF(msg);
    para_put_item(msg, "NEWCONN");
    para_put_item_int(msg, new_sock);
    para_put_item_int64(msg, (int64_t)curr_timestamp(NULL));
    msg_size=strlen(msg);
    if((res=ep_write(listener_ctl.ep,
                    listener_ctl.worker_slot[worker_idx],
                    msg, &msg_size)))
    {
        errlog("%s: listener ep_write worker(%d) fail, res:%d, "
                "abort, impossible",
                __func__, worker_idx, res);
        sock_close(new_sock);
    }

    return(0);
}/*}}}*/

static void _listener()
{/*{{{*/
#define CHECK_OPERATE_INFO_INTERVAL         1
    unsigned int quit=IS_FALSE;
    struct ep_active_slot *active_list;
    time_t last_operate=0;
    time_t now;
    int res;
    
    while(!quit)
    {
        res=ep_poll(listener_ctl.ep, &active_list);
        if(res)
        {
            errlog("%s: ep_poll fail, res:%d, impossible",
                    __func__, res);
            continue;
        }
        while(active_list)
        {
            switch(active_list->type)
            {
                case EP_SLOT_CONN:
                    _conn(active_list);
                    break;
                case EP_SLOT_ABORT:
                    errlog("%s: revoke slot(sock:%d,idx:%d) error(%d), "
                            "impossible",
                            __func__, active_list->_u._abort.sock,
                            active_list->slot_idx,
                            active_list->_u._abort.error);
                    break;
                default:
                    errlog("%s: listener poll type(%s) invalid, "
                            "impossible",
                            __func__, active_list->type);
                    break;
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }

        now=curr_timestamp(NULL);

        // check shutdown
        if(now - last_operate > CHECK_OPERATE_INFO_INTERVAL)
        {
            pthread_mutex_lock(&operate_info._listener_.lock); //lock
            quit=operate_info._listener_.quit;
            if(quit)
            {
                operate_info._listener_.state=THRD_SHUTDOWN;
            }
            pthread_mutex_unlock(&operate_info._listener_.lock); //unlock

            last_operate=now;
        }
    }
    errlog("%s: listener shutdown", __func__);
    DEEP_SLEEP;
}/*}}}*/

int listener_start(SOCKET *sock_pair)
{/*{{{*/
    int worker_idx;
    SOCKET sock;
    int res;

    // operate_info
    pthread_mutex_init(&operate_info._listener_.lock, NULL);
    operate_info._listener_.quit=IS_FALSE;
    operate_info._listener_.state=THRD_RUNNING;

    if((res=ep_create(&listener_ctl.ep)))
    {
        fprintf(stderr, "agent: listener(ep_create) fail, errcode:%d\n",
                res);
        return(-1);
    }

    // server
    if((sock=sock_listen(cnfg_info.server_port)) == INVALID_SOCKET)
    {
        fprintf(stderr, "agent: open server port fail\n");
        return(-1);
    }
    if((res=ep_slot_register(listener_ctl.ep, sock, IS_TRUE,
                    &listener_ctl.server_slot)))
    {
        fprintf(stderr,
                "agent: listener(ep_slot_register) server port fail, "
                "errcode:%d\n",
                res);
        return(-1);
    }
    fprintf(stderr, "agent: open server port(%d)\n",
            cnfg_info.server_port);

    // worker
    ALLOC_MULTI_DOERR(listener_ctl.worker_slot, EP_SLOT,
            cnfg_info.threads,
            fprintf(stderr,
                "agent: alloc listener_ctl.worker_slot fail\n");
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.threads; ++worker_idx)
    {
        if((res=ep_slot_register(listener_ctl.ep,
                        sock_pair[worker_idx*2], IS_FALSE,
                        &listener_ctl.worker_slot[worker_idx])))
        {
            fprintf(stderr,
                    "agent: listener(ep_register) worker(%d) fail, "
                    "errcode:%d\n",
                    worker_idx, res);
            return(-1);
        }
        ep_slot_set_wrap(listener_ctl.worker_slot[worker_idx]);
    }

    /*--- thread listener ---*/
    if(thrd_create((THRD_FUNC_T)_listener, (void *)NULL, THRD_DETACH,
                &listener_ctl.thrd_id))
    {
        fprintf(stderr, "agent: listener thrd create fail\n");
        return(-1);
    }

    return(0);
}/*}}}*/
